package handle

import (
	"context"
	"log"

	"github.com/rabbitmq/amqp091-go"
)

type rabbitHandle = func(engine *RabbitConn)

func NewRabbitEngine(handles ...rabbitHandle) *RabbitConn {
	engine := &RabbitConn{
		conn:      nil,
		channel:   nil,
		ConfigUrl: "",
	}
	for _, handle := range handles {
		handle(engine)
	}
	var err error
	engine.conn, err = amqp091.Dial(engine.ConfigUrl)
	if err != nil {
		panic(err)
	}
	engine.channel, err = engine.conn.Channel()
	if err != nil {
		panic(err)
	}
	err = engine.channel.Confirm(false)
	if err != nil {
		panic(err)
	}
	//engine.channel.NotifyPublish()
	return engine
}
func (r *RabbitConn) ListenConfirm(size int) {
	for ret := range r.channel.NotifyPublish(make(chan amqp091.Confirmation, size)) {
		if ret.Ack {
			log.Println("confirm: 消息发送成功", ret.DeliveryTag)
		} else {
			log.Println("confirm: 消息发送失败", ret.DeliveryTag)
		}
	}
}
func (r *RabbitConn) ChannelPublish(ctx context.Context, exchangeName string, routingKey string, data []byte) {
	if err := r.channel.PublishWithContext(ctx,
		exchangeName, // publish to an exchange
		routingKey,   // routing to 0 or more queues
		false,        // mandatory
		false,        // immediate
		amqp091.Publishing{
			Headers:         amqp091.Table{},
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            data,
			//persistent : 只要消息一达到队列, 就会立即写到文件来持久化
			//non-persistent: 它只会在 rabbitmq 存在内存压力的时候, 才会持久化到磁盘.
			DeliveryMode: amqp091.Transient, // 1=non-persistent, 2=persistent
			Priority:     0,                 // 0-9
			// a bunch of application/implementation-specific fields
		},
	); err != nil {
		panic(err)
	}

}
func (r *RabbitConn) CreateQuery(queueName string, table amqp091.Table) amqp091.Queue {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	q, err := r.channel.QueueDeclare(
		queueName,
		//是否持久化
		true,
		//是否自动删除
		false,
		//是否具有排他性
		false,
		//是否阻塞处理
		false,
		//额外的属性
		table,
	)
	if err != nil {
		log.Fatalf("Queue Declare: %s", err)
	}

	return q
}
func (r *RabbitConn) CreateExchange(exchange, exchangeType string, table amqp091.Table) {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	if err := r.channel.ExchangeDeclare(
		exchange,     // name
		exchangeType, // type  交换机模式 :direct,fanout, topic,headers
		false,        // durable
		false,        // auto-deleted
		false,        // internal
		false,        // noWait
		table,        // arguments
	); err != nil {
		log.Fatalf("Exchange Declare: %s", err)
	}

}

func (r *RabbitConn) Reliable(flag bool) {
	if flag {
		log.Printf("enabling publishing confirms.")
		if err := r.channel.Confirm(false); err != nil {
			log.Fatalf("Channel could not be put into confirm mode: %s", err)
		}
		confirms := r.channel.NotifyPublish(make(chan amqp091.Confirmation, 1))
		defer r.confirmOne(confirms)
	}

}
func (r *RabbitConn) confirmOne(confirms <-chan amqp091.Confirmation) {
	log.Printf("waiting for confirmation of one publishing")
	if confirmed := <-confirms; confirmed.Ack {
		log.Printf("confirmed delivery with delivery tag: %d", confirmed.DeliveryTag)
	} else {
		log.Printf("failed delivery of delivery tag: %d", confirmed.DeliveryTag)
	}
}

func (r *RabbitConn) Close() {
	r.channel.Close()
	r.conn.Close()
}
func (r *RabbitConn) GetConn() *amqp091.Connection {
	return r.conn
}
func (r *RabbitConn) GetChannel() *amqp091.Channel {
	return r.channel
}
